#ifndef __M_QUEUE_H__
#define __M_QUEUE_H__
#include "../mqcommon/mq_logger.hpp"
#include "../mqcommon/mq_helper.hpp"
#include "../mqcommon/mq_msg.pb.h"
#include <iostream>
#include <unordered_map>
#include <mutex>
#include <cassert>
#include <sstream>

namespace mymq
{
    struct MsgQueue
    {
        using ptr = std::shared_ptr<MsgQueue>;
        std::string name;
        bool durable;
        bool exclusive;
        bool auto_delete;
        google::protobuf::Map<std::string, std::string> args;

        MsgQueue(){}
        MsgQueue(const std::string& qname, bool qdurable, bool qexclusive, bool qauto_delete,
            const google::protobuf::Map<std::string, std::string>& qargs)
        : name(qname), durable(qdurable), exclusive(qexclusive), auto_delete(qauto_delete), args(qargs)
        {}

        // 参数持久化存储格式：key=val&key=val&key=val
        // 内部解析字符串，将参数存储到内部成员中
        void setArgs(const std::string& _args)
        {
            std::vector<std::string> kvs;
            StrHelper::split(_args, "&", kvs);
            // 得到 key=val...
            for (auto& str : kvs)
            {
                int pos = str.find("=");
                std::string key = str.substr(0, pos);
                std::string val = str.substr(pos + 1);
                args[key] = val;
            }
        }   
        // 将内部参数进行序列化，返回一个字符串
        std::string getArgs()
        {
            std::string ret;
            for (auto it = args.begin(); it != args.end(); ++it)
            {
                ret += it->first + "=" + it->second + "&";
            }
            return ret;
        }
    };

    using QueueMap = std::unordered_map<std::string, MsgQueue::ptr>;
    class MsgQueueMapper
    {
    public:
        MsgQueueMapper(const std::string& dbfile)
        : _sql_helper(dbfile)
        {
            // 确保创建数据库之前，父级目录存在
            std::string par_path = FileHelper::parentDirectory(dbfile);
            FileHelper::createDirectory(par_path);
            assert(_sql_helper.open()); // 打开数据库，没有就创建
            createTable();
        }

        // 创建表
        void createTable()
        {
            std::stringstream sql;
            sql << "create table if not exists queue_table(";
            sql << "name varchar(32) primary key, ";
            sql << "durable int, ";
            sql << "exclusive int, ";
            sql << "auto_delete int, ";
            sql << "args varchar(128));";

            bool ret = _sql_helper.exec(sql.str(), nullptr, nullptr);
            if (ret == false)
            {
                ELOG("创建队列数据表失败！");
                abort();
            }
        }

        // 删除表
        void removeTable()
        {
            std::string sql = "drop table if exists queue_table;";

            bool ret = _sql_helper.exec(sql, nullptr, nullptr);
            if (ret == false)
            {
                ELOG("删除队列数据表失败！");
                abort();
            }
        }

        // 插入队列数据
        bool insert(MsgQueue::ptr& queue)
        {
            // insert into queue_table values('q1', t, f, f, "key1=val1&key2=val2&");
            std::stringstream sql;
            sql << "insert into queue_table values(";
            sql << "'" << queue->name << "', ";
            sql << queue->durable << ", ";
            sql << queue->exclusive << ", ";
            sql << queue->auto_delete << ", ";
            sql << "'" << queue->getArgs() << "');";

            return _sql_helper.exec(sql.str(), nullptr, nullptr);
        }

        // 删除队列数据
        void remove(const std::string& name)
        {
            // delete from queue_table where name='q1';
            std::stringstream sql;
            sql << "delete from queue_table where name=";
            sql << "'" << name << "';";
            _sql_helper.exec(sql.str(), nullptr, nullptr);
        }

        // 取出所有队列数据 队列名-队列对象
        QueueMap recovery()
        {
            // select * from queue_table;
            std::string sql = "select * from queue_table;";
            QueueMap result;
            _sql_helper.exec(sql, selectCallBack, &result);
            return result;
        }
    private:
        static int selectCallBack(void* args, int numcol, char** row, char** fields)
        {
            QueueMap* result = (QueueMap*)args;
            MsgQueue::ptr mqp = std::make_shared<MsgQueue>();
            mqp->name = row[0];
            mqp->durable = (bool)std::stoi(row[1]);
            mqp->exclusive = (bool)std::stoi(row[2]);
            mqp->auto_delete = (bool)std::stoi(row[3]);
            if (row[4]) mqp->setArgs(row[4]);

            result->insert({mqp->name, mqp});
            return 0;
        }
    private:
        SqliteHelper _sql_helper;
    };

    class MsgQueueManager
    {
    public:
        using ptr = std::shared_ptr<MsgQueueManager>;
        MsgQueueManager(const std::string& dbfile)
        : _mapper(dbfile)
        {
            _msg_queues = _mapper.recovery();
        }

        bool declareQueue(const std::string& qname, bool qdurable, bool qexclusive, bool qauto_delete,
            const google::protobuf::Map<std::string, std::string>& qargs)
        {
            // 队列存在就 OK，不存在就创建
            std::unique_lock<std::mutex> lock(_mutex);

            // 判断队列是否存在
            if (_msg_queues.find(qname) != _msg_queues.end()) return true;

            // 队列不存在
            MsgQueue::ptr mqp = std::make_shared<MsgQueue>(qname, qdurable, qexclusive, qauto_delete, qargs);
            if (qdurable == true)
            {
                bool ret = _mapper.insert(mqp);
                if (ret == false) return false;
            }
            _msg_queues.insert(std::make_pair(qname, mqp));
            return true;
        }

        void deleteQueue(const std::string& name)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            // 判断队列是否存在
            if (_msg_queues.find(name) == _msg_queues.end()) return;

            if (_msg_queues[name]->durable == true)
                _mapper.remove(name);
            _msg_queues.erase(name);
        }

        MsgQueue::ptr selectQueue(const std::string& name)
        {
            std::unique_lock<std::mutex> lock(_mutex);

            if (_msg_queues.find(name) == _msg_queues.end()) return MsgQueue::ptr();
            return _msg_queues[name];
        }

        QueueMap getAllQueues()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _msg_queues;
        }

        bool exists(const std::string& name)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _msg_queues.find(name) != _msg_queues.end();
        }

        size_t size()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            return _msg_queues.size();
        }

        void clear()
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _mapper.removeTable();
            _msg_queues.clear();
        }

    private:
        std::mutex _mutex;
        MsgQueueMapper _mapper;
        QueueMap _msg_queues;
    };
}
#endif